package com.meteor.raft.core;

import com.alibaba.fastjson.JSON;
import com.meteor.raft.constant.CommonConstant;
import com.meteor.raft.executor.GlobalExecutor;
import com.meteor.raft.healthcheck.HealthCheckService;
import com.meteor.raft.masterelection.ElectionService;
import com.meteor.raft.misc.HttpClient;
import com.meteor.raft.misc.UrlUtil;
import com.meteor.raft.persistent.RaftPeer;
import com.ning.http.client.AsyncCompletionHandler;
import com.ning.http.client.Response;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.HttpURLConnection;
import java.util.concurrent.atomic.AtomicLong;

/**
 * @author gang.wang1903@xiaobao100.cn
 * @Date: 2019/7/15 8:07 PM
 * raft核心实现类
 */
@Component
@Slf4j
public class RaftCore {


    @Autowired
    private RaftPeerManage raftPeerManage;

    @Autowired
    private ElectionService electionService;

    @Autowired
    private HealthCheckService healthCheckService;



    /**
     * 核心类初始化:包括心跳检测，master选举
     * @throws Exception
     */
    @PostConstruct
    public void init() throws Exception{
        GlobalExecutor.registerMasterElection(new MasterElection());
        GlobalExecutor.registerHeartbeat(new HeartBeat());
        GlobalExecutor.registeIsReady(new meteorIsReady());
    }


    /**
     * master选举
     */
    public class MasterElection implements Runnable {
        @Override
        public void run() {
            try {

                if (!raftPeerManage.isReady()) {
                    return;
                }
                RaftPeer local = raftPeerManage.local();
                local.leaderDueMs -= GlobalExecutor.HEARTBEAT_INTERVAL_MS;

                if (local.leaderDueMs > 0) {
                    return;
                }
                // reset timeout
                local.resetLeaderDue();
                local.resetHeartbeatDue();
                log.info("[RAFT] start  master election");
                electionService.sendVote();
            } catch (Exception e) {
                log.error("[RAFT] error  master election {}", e);
            }

        }
    }


    /**
     * 用于节点心跳检查
     */
    public class HeartBeat implements Runnable {
        @Override
        public void run() {
            try {

                if (!raftPeerManage.isReady()) {
                    return;
                }

                RaftPeer local = raftPeerManage.local();
                /*local.heartbeatDueMs -= GlobalExecutor.TICK_PERIOD_MS;
                if (local.heartbeatDueMs > 0) {
                    return;
                }
                local.resetHeartbeatDue();*/
                healthCheckService.sendBeat();
            } catch (Exception e) {
                log.error("[RAFT] error  sending beat {}", e);
            }

        }
    }



    /**
     * 所有节点是否准备
     */
    public class meteorIsReady implements Runnable {

        private AtomicLong localTerm = new AtomicLong(0L);

        @Override
        public void run() {
            try {
                /**
                 * 如果所有节点准备好了直接返回
                 */
                if(raftPeerManage.isReady()){
                    return;
                }
                if(raftPeerManage.isStandalone()){
                    raftPeerManage.setReady(true);
                    return;
                }

                for (final String server : raftPeerManage.allServersWithoutMySelf()) {
                    final String url = UrlUtil.buildURL(server, CommonConstant.API_ISREADY);
                    try {
                        HttpClient.asyncHttpPost(url, null, null, new AsyncCompletionHandler<Integer>() {
                            @Override
                            public Integer onCompleted(Response response) throws Exception {
                                if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                                    log.error("RAFT vote failed: {}, url: {}", response.getResponseBody(), url);
                                    return 1;
                                }

                                boolean result = Boolean.parseBoolean(response.getResponseBody());

                                if(result){
                                    localTerm.incrementAndGet();
                                }

                                return 0;
                            }
                        });
                    } catch (Exception e) {
                        log.warn("error while ready: {}", server);
                    }
                }
                if(raftPeerManage.getPeers().keySet().size() == localTerm.intValue()){
                    raftPeerManage.setStandalone(true);
                }
            } catch (Exception e) {
                log.error("[RAFT] error  ready {}", e);
            }

        }
    }
}
